package com.xiguthings.xiniu.iot.device.state.controller;

import javax.annotation.PostConstruct;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import com.xiguthings.xiniu.iot.device.state.jedis.DeviceDataBuffer;
import com.xiguthings.xiniu.iot.device.state.kafka.ConsumerRunnable;
import com.xiguthings.xiniu.iot.device.state.kafka.DataHandlerPool;
import com.xiguthings.xiniu.iot.device.state.kafka.KafkaClient;

@Component
public class Action {
	@Autowired
	private DeviceDataBuffer deviceDataBuffer;
	// kafka客户端线程池的大小
	@Value("${kafka.consumer.pool.thread.count}")
	private int kafkaConsumerPoolThreadCount;

	@Autowired
	private KafkaClient kafkaClient;
	@Autowired
	DataHandlerPool dataHandlerPool;

	@PostConstruct
	// @Override
	public void run() throws Exception {
		deviceDataBuffer.flush();
		for (int i = 0; i < kafkaConsumerPoolThreadCount; i++) {
			submintOneKafkaThread();
		}
	}

	/**
	 * 提交一个kafka消费者线程给线程池执行
	 */
	private void submintOneKafkaThread() {
		DataCenter dataCenter = new DataCenter();// 每个线程上都创建一个信息中心
		kafkaClient.setDataCenter(dataCenter);// kafka客户端需要指定一个信息中心
		ConsumerRunnable consumerRunnable = kafkaClient.getConsumerRunnable();
		// 将线程交给线程池执行
		dataHandlerPool.submitConsumerPool(consumerRunnable);// 提交消费者线程任务到线程池（开始运行线程）
	}

}
